# -*- coding: utf-8 -*-
from datetime import timedelta
from utils.operators.spark_sql_operator import SparkSqlOperator
from jms.ods.tab.tab_barscan_taking import jms_ods__tab_barscan_taking
from jms.ods.tab.tab_barscan_warehousing import jms_ods__tab_barscan_warehousing
from jms.ods.tab.tab_barscan_unloading import jms_ods__tab_barscan_unloading
from jms.ods.tab.tab_barscan_loading import jms_ods__tab_barscan_loading
from jms.ods.tab.tab_barscan_bagging import jms_ods__tab_barscan_bagging
from jms.ods.tab.tab_barscan_sitesend import jms_ods__tab_barscan_sitesend
from jms.ods.tab.tab_barscan_centersend import jms_ods__tab_barscan_centersend
from jms.ods.tab.tab_barscan_sitearrival import jms_ods__tab_barscan_sitearrival
from jms.ods.tab.tab_barscan_centerarrival import jms_ods__tab_barscan_centerarrival
from jms.ods.tab.tab_barscan_deliver import jms_ods__tab_barscan_deliver
from jms.ods.tab.tab_barscan_sign import jms_ods__tab_barscan_sign
from jms.ods.tab.tab_barscan_collect import jms_ods__tab_barscan_collect
from jms.ods.tms.yl_tmsnew_branch_shipment_stop import jms_ods__yl_tmsnew_branch_shipment_stop
from jms.ods.tms.yl_tmsnew_tms_shipment_stop import jms_ods__yl_tmsnew_tms_shipment_stop
from jms.ods.lmdm.yl_lmdm_sys_network import jms_ods__yl_lmdm_sys_network

out_days = 15  # 输出最近 15 天的分区
in_days = out_days + 15  # 读入最近 15 + 8 天的扫描数据

jms_dm__dm_waybill_trace_dt = SparkSqlOperator(
    task_id='jms_dm__dm_waybill_trace_dt',
    email=['lukunming@jtexpress.com','yl_bigdata@yl-scm.com'],
    name='jms_dm__dm_waybill_trace_dt_{{ execution_date | cst_ds }}',
    email_on_retry=True,
    sql='jms/dm/dm_waybill_trace_dt/execute.sql',
    yarn_queue='pro',
    driver_memory='12G',
    driver_cores=8,
    executor_memory='16G',
    num_executors=70,
    executor_cores=8,
    pool_slots=27,
    conf={'spark.executor.memoryOverhead'                            : '5G',
          'spark.dynamicAllocation.enabled'                          : 'true',
          'spark.shuffle.service.enabled'                            : 'true',
          'spark.dynamicAllocation.maxExecutors'                     : 120,
          'spark.dynamicAllocation.cachedExecutorIdleTimeout'        : 20,
          'spark.sql.shuffle.partitions'                             : in_days * 100,
          'spark.default.paralleism': '2000',  # spark.default.parallelism只有在处理RDD时有效.
          },
    hiveconf={'hive.exec.dynamic.partition'             : 'true',
              'hive.exec.dynamic.partition.mode'        : 'nonstrict',
              'hive.exec.max.dynamic.partitions'        : out_days,
              'hive.exec.max.dynamic.partitions.pernode': out_days,
              },
    params={'out_days': out_days,
            'in_days': in_days,
            },
    execution_timeout=timedelta(minutes=70),
)

jms_dm__dm_waybill_trace_dt << [
    jms_ods__tab_barscan_taking,
    jms_ods__tab_barscan_warehousing,
    jms_ods__tab_barscan_unloading,
    jms_ods__tab_barscan_loading,
    jms_ods__tab_barscan_bagging,
    jms_ods__tab_barscan_sitesend,
    jms_ods__tab_barscan_centersend,
    jms_ods__tab_barscan_sitearrival,
    jms_ods__tab_barscan_centerarrival,
    jms_ods__tab_barscan_deliver,
    jms_ods__tab_barscan_sign,
    jms_ods__tab_barscan_collect,
    jms_ods__yl_tmsnew_branch_shipment_stop,
    jms_ods__yl_tmsnew_tms_shipment_stop,
    jms_ods__yl_lmdm_sys_network,
]
